package com.ibeetl.admin.aod.service.excel.processor.data;

import com.ibeetl.admin.aod.entity.AodIndustryData;
import com.ibeetl.admin.aod.service.excel.task.InsertDataTask;
import com.ibeetl.admin.aod.util.ThreadPoolFactory;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.FutureTask;

/**
 * 待重构 TODO
 */
@Component
public class BatchInsertDataProcessor {

    protected final Logger log = LoggerFactory.getLogger(this.getClass());

    protected static ExecutorService executorService = ThreadPoolFactory.getThreadPool("insertThread", 10);

    @Autowired
    protected Environment environment;

    protected final int defaultSize = 1000;

    private Connection getConnection() throws Exception {
        String url = environment.getProperty("spring.datasource.url");
        String user = environment.getProperty("spring.datasource.username");
        String password = environment.getProperty("spring.datasource.password");
        Class.forName("com.mysql.jdbc.Driver");
        Connection conn = DriverManager.getConnection(url, user, password);
        conn.setAutoCommit(false);
        return conn;
    }

    public boolean insertMapData(Map<String,List<AodIndustryData>> aodIndustryDataMap) {

        boolean result = true;

        log.info("进入入库主入口");
        if (null != aodIndustryDataMap && aodIndustryDataMap.size() > 0) {

            List<FutureTask<Integer>> futureTaskList = new ArrayList<>();
            List<Connection> connectionList = new ArrayList<>();

            String insertSql = environment.getProperty("aod.insertPrefixSql");

            for (Map.Entry<String,List<AodIndustryData>> entry : aodIndustryDataMap.entrySet()) {
                List<AodIndustryData> aodIndustryDataList = entry.getValue();
                log.info("行业表名{} 对应需要插入的数据条数为{}", entry.getKey(), aodIndustryDataList.size());
                // 尝试10个线程平分数据 插入任务
                // 单个线程处理的数据量
                int size = aodIndustryDataList.size();
                int threadDataNum = size;
                int threadNum = 1;
                // 少于1000的数据,直接单线程插入
                if (defaultSize < size) {
                    threadDataNum = (size % 10) == 0 ? (size / 10) : (size / 10) + 1;
                    threadNum = (size % threadDataNum) == 0 ? (size / threadDataNum) : (size / threadDataNum) + 1;
                }

                int counter = 0;

                for (int i=0;i<threadNum;i++) {
                    int fromIndex = i * threadDataNum;
                    int endIndex = (counter + threadDataNum) > aodIndustryDataList.size() ? aodIndustryDataList.size() : (counter + threadDataNum);

                    List<AodIndustryData> tmpAodIndustryDataList = aodIndustryDataList.subList(fromIndex, endIndex);

                    try {
                        Connection connection = getConnection();
                        connectionList.add(connection);
                        FutureTask futureTask = new FutureTask(new InsertDataTask(tmpAodIndustryDataList, connection, String.format(insertSql, entry.getKey())));
                        futureTaskList.add(futureTask);
                        executorService.submit(futureTask);
                    } catch (Exception e) {
                        log.info("获取连接失败",e);
                    }
                    counter += endIndex - fromIndex;
                }

            }

            result = judeResult(futureTaskList);

            postProcess(result, connectionList);

        }
        return result;

    }

    public boolean judeResult(List<FutureTask<Integer>> futureTaskList) {
        boolean result = true;
        for (FutureTask futureTask : futureTaskList) {
            try {
                futureTask.get();
            } catch (Exception e) {
                log.info("插入数据失败",e);
                result = false;
            }
        }
        return result;
    }

    public void postProcess(boolean noNeedRollback, List<Connection> connectionList) {
        if (!noNeedRollback) {
            // 只要其中一个抛出异常 直接所有connection全部执行回滚
            // 此操作应该抽到 线程任务中去 TODO
            if (CollectionUtils.isNotEmpty(connectionList)) {
                for (Connection connection : connectionList) {
                    try {
                        connection.rollback();
                        connection.close();
                    } catch (Exception e1) {
                        log.info("回滚事务失败",e1);
                    }
                }
            }
        } else {
            for (Connection connection : connectionList) {
                try {
                    connection.commit();
                    connection.close();
                } catch (Exception e1) {
                    log.info("提交事务失败",e1);
                }
            }
        }
    }
    //
    public void insertData(List<AodIndustryData> aodIndustryDataList, String tableName) {
        if (CollectionUtils.isNotEmpty(aodIndustryDataList)) {
            String insertSql = environment.getProperty("aod.insertPrefixSql");
            // 尝试10个线程平分数据 插入任务
            // 单个线程处理的数据量
            int size = aodIndustryDataList.size();
            int threadDataNum = size;
            int threadNum = 1;
            if (defaultSize < size) {
                threadDataNum = (size % 10) == 0 ? (size / 10) : (size / 10) + 1;
                threadNum = (size % threadDataNum) == 0 ? (size / threadDataNum) : (size / threadDataNum) + 1;
            }

            int counter = 0;

            List<FutureTask<Integer>> futureTaskList = new ArrayList<>();
            List<Connection> connectionList = new ArrayList<>();

            for (int i=0;i<threadNum;i++) {
                int fromIndex = i * threadDataNum;
                int endIndex = (counter + threadDataNum) > aodIndustryDataList.size() ? aodIndustryDataList.size() : (counter + threadDataNum);

                List<AodIndustryData> tmpAodIndustryDataList = aodIndustryDataList.subList(fromIndex, endIndex);

                try {
                    Connection connection = getConnection();
                    connectionList.add(connection);
                    FutureTask futureTask = new FutureTask(new InsertDataTask(tmpAodIndustryDataList, connection, String.format(insertSql, insertSql)));
                    futureTaskList.add(futureTask);
                    executorService.submit(futureTask);
                } catch (Exception e) {
                    log.info("获取连接失败",e);
                }
                counter += endIndex - fromIndex;
            }

            boolean needRollback = judeResult(futureTaskList);

            postProcess(needRollback, connectionList);

        }
    }

}
